/*!
 redis stream consumer deamon

*/
use crate::shutdown::Shutdown;
use crate::*;

use rabbitmq::client::ChannelClient;
// use chrono::prelude::*;
use redis::streams::{StreamId, StreamKey, StreamReadReply};
use redis::Value;
use redis::{streams::StreamMaxlen, AsyncCommands, RedisResult};
use tokio::sync::mpsc::Sender;

use log::Level::{Error, Info};
#[cfg(feature = "rotatelog")]
use log::{debug, error, info, log as event, warn};
#[cfg(feature = "tracelog")]
use tracing::{debug, error, event, info};

use tokio::sync::mpsc;

pub const TOPIC: &str = "gfs-sync";
const NEW_FROM_NOW: &str = "$";
const FROM_LAST_START: &str = ">";
const FROM_FIRST_START: &str = "0-0";

pub(crate) struct MqConsumer {
    pub rmq_client: Arc<ChannelClient>,
    pub shutdown: shutdown::Shutdown,
    /// Not used directly. Instead, when `Handler` is dropped...?
    pub _shutdown_complete: mpsc::Sender<()>,
    pub shared: Arc<ShareComm>,
}

impl MqConsumer {
    //  架构调整优化！
    //  等待删除  ！！
    //
    /* 
    pub async fn brun(&mut self, group: &str, cname: &str) -> Result<()> {
        if self.shared.is_master() {
            return Ok(());
        }

        if self.shared.mq_server.starts_with("redis") {
            let client = redis::Client::open(self.shared.mq_server.as_str())
                .expect("failed to create redis client");
            let mqc = MqRedis {
                server_name: self.shared.mq_server.clone(),
                client,
            };
            if let Err(err) = mqc
                .run(&self.shared.tx, TOPIC, group, cname, &mut self.shutdown)
                .await
            {
                bail!("NO")
            }
        } else {
            //
            let mut retry = true;
            while retry {
                let mqc = MqRabbit {
                    server_name: self.shared.mq_server.clone(),
                    channel_client: Arc::clone(&self.rmq_client),
                };

                if let Err(err) = mqc
                    .run(&self.shared.tx, TOPIC, cname, &mut self.shutdown)
                    .await
                {
                    match err {
                        MyError::SubscribeErr(err) => {
                            error!(
                                "subscribe error, mybe format is error,frist will exit {}",
                                err
                            );
                        }
                        MyError::Terminated(err) => {
                            error!(
                                "received is None, mybe server is down, should to reconnect {}",
                                err
                            );
                        }
                        MyError::ConnectRefused(err) => {
                            error!("server down, will try to reconnect {}", err);
                        }
                        _ => {
                            error!("unkown error, first will exit: {:#?}", err);
                            retry = false;
                        }
                    }
                } else {
                    retry = false;
                }
                if retry {
                    // 不确认sleep是否相应ctrl-c，如果不响应，需要select监听
                    let mut ticker = tokio::time::interval(Duration::from_secs(10));
                    ticker.tick().await;
                    tokio::select! {
                        _ = ticker.tick() => {  info!("try to connect server again");    },
                       _ = self.shutdown.recv() => {
                            break
                       },
                    }
                }
            }
        }
        event!(target:"shutdown", loglevel(Error), "MQ Consummer thread  is shutdown, OK");
        Ok(())
    }
 */
    pub async fn run(&mut self, group: &str, cname: &str) -> Result<()> {
        if self.shared.is_master() {
            return Ok(());
        }
        let actor = if self.shared.mq_server.starts_with("redis") {
            let client = redis::Client::open(self.shared.mq_server.as_str())
                .expect("failed to create redis client");
            let mqc = MqRedis {
                server_name: self.shared.mq_server.clone(),
                client,
            };
            MqActor::RedisMq(mqc)
        } else {            
            let mqc = MqRabbit {
                server_name: self.shared.mq_server.clone(),
                slave_port: self.shared.port,
                channel_client: Arc::clone(&self.rmq_client),
            };
            MqActor::RabbitMq(mqc)
        };

        if let Err(err) = actor
            .run(&self.shared.tx, TOPIC, group, cname, &mut self.shutdown)
            .await
        {
            error!("{:#?}", err);
        }
        event!(target:"shutdown", loglevel(Error), "MQ Consummer thread  is shutdown, OK");
        Ok(())
    }
}

/*
监听MQ，接收来自master发出的新文件消息，并把消息传递给负责同步的worker线程。
 - 因为没有很多的需要MQ的场景，故而这里直接通过不同分支，完成了不同MQ类型的逻辑执行。
 - 实际更合理的架构应该是外部根据环境配置，生成MqActor，通过shared传递到这里，这里只需要执行MqActor::listen即可，无需鉴定具体的MQ逻辑。
*/

/**
从指定stream key中接收发给group的消息
 - 采用异步连接：let mut con = client.get_async_connection().await?;
 - 设置BLOCK=0：StreamReadOptions.block(0)
 - 消费完成后需要XACK确认。
 - 我理解的逻辑：XREAD处理的都是监听开始后的新消息，如果想读取或处理之前的消息，需要使用XRANGE系列命令（此时没有group，consumer）。

*/
async fn redis_next_mq(
    client: &redis::Client,
    group: &str,
    cname: &str,
    streamkey: &str,
) -> RedisResult<StreamReadReply> {
    let mut con = client.get_async_connection().await?;
    // let mut con = client.get_connection().expect("conn");
    let opts = redis::streams::StreamReadOptions::default()
        .block(0)
        .count(3)
        .group(group, cname);

    //可以同时获取多个key, .xread_options(&[TOPIC_A, TOPIC_B], &[rops.clone(), rops.clone()], &opts)
    let srr: StreamReadReply = con
        .xread_options(&[streamkey], &[FROM_LAST_START], &opts)
        .await
        .expect("records");
    redis::RedisResult::Ok(srr)
}

/**
删除指定的组中的消费者，用于消费者系统下线时的操作。

*/
async fn redis_delete_consumer(client: &redis::Client, skey: &str, group: &str, cname: &str) {
    // xgroup_delconsumer(key: K, group: G, consumer: C) -> RedisResult<RV>
    debug!(
        "yes, should remove consumer:{} from group:{} for stream key:{}",
        cname, group, skey
    );

    let mut con = client.get_async_connection().await.unwrap();
    let _r: isize = con.xgroup_delconsumer(skey, group, cname).await.unwrap();
}

/**
固定队列最大长度规则下添加新消息。
 - skey: &str
 - key: &str
 - val: &str

 实际增加内容可以不限数量的K-V值对：xadd(skey, "*", &[(key, val), ("sys", "系统加入，忽略"), ... ])

 处理完成的消息数超过规定值，将删除（标记删除？）最旧消息，遵守最大要求。
 - 这个行为不是准确实时的，而是异步的，大致达到2倍最大数时，按旧序删除最大数多出的消息。
 - 系统为保证空间，删除行为不管是否ACK，未消费消息也会被删除。

 */
pub async fn redis_message_to_mq(
    client: &redis::Client,
    skey: &str,
    key: &str,
    val: &str,
) -> RedisResult<()> {
    let mut con = client.get_async_connection().await?;
    let maxlen = StreamMaxlen::Approx(100);
    let _r: String = con.xadd_maxlen(skey, maxlen, "*", &[(key, val)]).await?;
    return RedisResult::Ok(());
}
/**
监听来自订阅topic的rabbitMQ消息，解析为新文档信息后通过tx提交给负责同步的worker线程。
 - 这里有一个投机：没有让MqRabbit去维护一个复用的rabbitMQ client
 - 返回了几种不同类型的错误，便于启动线程判断是否需要重试（服务器可能down了，需要重联）
 */
async fn rabbitmq_consumer(
    tx: &Sender<Task>,
    mq_server: &str,
    slave_port:u16,
    topic: &str,
    shutdown: &mut Shutdown,
) -> std::result::Result<(), MyError> {
    let topics: Vec<String> = vec![topic.to_string()];
    let client = match rabbitmq::client::connect(mq_server, "token", slave_port ).await {
        Ok(client) => client,
        Err(err) => return Err(MyError::ConnectRefused(format!("{:#?}", err))),
    };
    debug!("begin to listen: {:#?}", &topics);

    let mut subscriber = match client.subscribe(topics).await {
        Ok(subscriber) => subscriber,
        Err(err) => return Err(MyError::SubscribeErr(format!("{:#?}", err))),
    };

    event!(target:"startup", loglevel(Info), "Rabbit-Consummer thread  is startup, OK");

    while !shutdown.is_shutdown() {
        tokio::select! {
            stream = subscriber.next_message() => {
                match stream {
                    Ok(msg) => {
                        if let Some(msg) =  msg {
                            let resp:GfsResponse  = serde_json::from_slice(&msg.content).unwrap_or( GfsResponse { name: "".to_string(), ihash:0, idx:0, code: -1 }  );

                            if resp.code != -1 {
                                if let Err(_) = tx.send(Task::Sync(resp)).await{
                                    error!("receiver dropped, 不可能发生！！");
                                }
                            }
                        }else{
                            error!("the subscription has been terminated. mybe server is down!");
                            return Err(MyError::Terminated("the subscription has been terminated. mybe server is down!".to_string()))
                        }
                    },
                    Err(err) =>{
                        debug!("next_message error! {:#?}, if should reconnect?", err);
                        // tokio::time::sleep( std::time::Duration::from_secs(2)).await;
                    }
                }
            },
            _ = shutdown.recv() => {
                break
            },
        }
    }
    Ok(())
}

/**
监听来自redis的MQ消息，解析为新文档信息后通过tx提交给负责同步的worker线程。
 - 这里实际为何保持和MqRabbit的一致，提供了此函数，否则可以让入MqRedis的listen中，通过MqRedis维护复用的client
 - 不同于rabbitMQ, redis的client应该具备自动重连的能力，但为了保持代码的一致性，仍然提供了类似的结构和机制。
 */
async fn redismq_consumer(
    tx: &Sender<Task>,
    mq_server: &str,
    topic: &str,
    group: &str,
    cname: &str,
    shutdown: &mut Shutdown,
) -> std::result::Result<(), MyError> {
    //    let topic = TOPIC;
    let client = match redis::Client::open(mq_server) {
        Ok(client) => client,
        Err(err) => return Err(MyError::ConnectRefused(format!("{:#?}", err))),
    }; //.expect("redis client failed");

    let mut con = match client.get_async_connection().await {
        Ok(client) => client,
        Err(err) => return Err(MyError::ConnectRefused(format!("{:#?}", err))),
    }; //  .expect("connect server error");

    // 先要创建group
    // only new from now()
    let created: redis::RedisResult<()> =
        con.xgroup_create_mkstream(topic, group, NEW_FROM_NOW).await;
    if let Err(e) = created {
        error!("Group already exists: {:?}", e)
    } else {
        info!(
            "yes, jion group = {} , custmername = {},  stream key = {}",
            group, cname, topic
        )
    }

    event!(target:"startup", loglevel(Info), "Redis-Consummer thread  is startup, OK");

    while !shutdown.is_shutdown() {
        tokio::select! {
            stream = redis_next_mq(&client, group, cname, topic) => {
                match stream {
                redis::RedisResult::Ok(msg) => {
                    let mut ids_done: Vec<String> = vec![];
                    for StreamKey { key, ids } in msg.keys {
                        // println!("Stream {}", key);
                        for StreamId { id, map } in ids {
                            //let ct = parse_id_to_datetime(id.as_str());
                            //println!("\tID {} ({:#?}), content map counts={}", id, ct, map.len());
                            for (kid, s) in map {
                                if let Value::Data(bytes) = s {
                                    error!("gfs_copy id = {}", kid);
                                    let resp:GfsResponse  = serde_json::from_slice(&bytes).unwrap_or( GfsResponse { name: "".to_string(), ihash:0, idx:0, code: -1 }  );

                                    if resp.code != -1 {
                                        if let Err(_) = tx.send(Task::Sync(resp)).await{
                                            error!("receiver dropped, 不可能发生！！");
                                        }else{
                                            ids_done.push(id.clone());
                                        }
                                    }
                                } else {
                                    error!("Weird data")
                                }
                            }
                        }
                        // ACK 确认消费成功！ 问题：
                        // 1. 消费者读取了，却没有ACK，可能会再次被消费！
                        // 2. 如果没有ACK，再次消费的条件时什么 ？会不会乱序 ？是否考虑幂等 ？

                        if ids_done.len() > 0 {
                            let _:isize = con.xack(key, group, &ids_done).await.expect("ack");
                        }
                    }
                },
                Err(err)=> {
                    error!("read message error! {:#?}, if should reconnect?", err);
                    return Err(MyError::Terminated(format!("read mq error: {:#?}. mybe server is down!", err)))
                },
            }

            },
            _ = shutdown.recv() => {
                 break
                },

        }
    }

    redis_delete_consumer(&client, TOPIC, group, cname).await;
    Ok(())
}

// pub struct MqNone {}
/**
使用redis作为MQ服务的实现结构
 */

pub struct MqRedis {
    pub client: redis::Client,
    pub server_name: String,
}
/**
使用内置rabbitmq作为MQ服务的实现结构
 - 没有维护rabbitmq::client::Client,因为场景没有这个必要，同时此Client的实现也不支持Copy，给编码带来挑战。
 */
pub struct MqRabbit {
    pub server_name: String,
    pub slave_port: u16,
    pub channel_client: Arc<rabbitmq::client::ChannelClient>,
}

impl MqRabbit {
    async fn publish(&self, topic: &str, _key: &str, val: &str) -> Result<()> {
        debug!("topic={}, val={}", topic, val);
        if let Ok(()) = self.channel_client.publish(topic, val).await {
            return Ok(());
        }
        bail!("error")
    }
    async fn run(
        &self,
        tx: &Sender<Task>,
        topic: &str,
        cname: &str,
        shutdown: &mut Shutdown,
    ) -> std::result::Result<(), MyError> {
        debug!("this is Rabbit for {}", cname);
        //   rabbitmq_consumer(tx, &self.server_name, topic, shutdown).await
        let mut retry = true;
        while retry {
            if let Err(err) = rabbitmq_consumer(tx, &self.server_name, self.slave_port, topic, shutdown).await {
                match err {
                    MyError::SubscribeErr(err) => {
                        error!(
                            "subscribe error, mybe format is error,frist will exit {}",
                            err
                        );
                    }
                    MyError::Terminated(err) => {
                        error!(
                            "received is None, mybe server is down, should to reconnect {}",
                            err
                        );
                    }
                    MyError::ConnectRefused(err) => {
                        error!("server down, will try to reconnect {}", err);
                    }
                    _ => {
                        error!("unkown error, first will exit: {:#?}", err);
                        retry = false;
                    }
                }
            } else {
                retry = false;
            }
            if retry {
                let mut ticker = tokio::time::interval(Duration::from_secs(10));
                ticker.tick().await;
                tokio::select! {
                    _ = ticker.tick() => {  info!("try to connect server again");    },
                   _ = shutdown.recv() => {
                        break
                   },
                }
            }
        }
        Ok(())
    }
}

impl MqRedis {
    async fn run(
        &self,
        tx: &Sender<Task>,
        topic: &str,
        group: &str,
        cname: &str,
        shutdown: &mut Shutdown,
    ) -> std::result::Result<(), MyError> {
        redismq_consumer(tx, &self.server_name, topic, group, cname, shutdown).await
    }
    async fn publish(&self, topic: &str, key: &str, val: &str) -> Result<()> {
        let client = match redis::Client::open(self.server_name.as_str()) {
            Ok(client) => client,
            Err(err) =>{
                error!("{}", err);
                bail!("redis error")},
        }; 

        if let Err(err) = redis_message_to_mq(&client, topic, key, val).await {
            error!("send sync msg to redis error:  {:#?}", err);
            bail!(format!("{:#?}", err))
        }
        Ok(())
    }
}

/**
通过MqActor，实现动态执行不同对象结构行为的目的！
 - 这应该是rust典型的实现手段
 - 可以对比trait + Box<dyn trait> 的传统面向对象的实现方法 
 */

pub enum MqActor {
    RedisMq(MqRedis),
    RabbitMq(MqRabbit),
    NoneMq(),
}

impl MqActor {
    pub(crate) async fn run(
        &self,
        tx: &Sender<Task>,
        topic: &str,
        group: &str,
        cname: &str,
        shutdown: &mut Shutdown,
    ) -> std::result::Result<(), MyError> {
        use crate::MqActor::*;
        match self {
            RedisMq(actor) => actor.run(tx, topic, group, cname, shutdown).await,
            RabbitMq(actor) => actor.run(tx, topic, cname, shutdown).await,
            _ => {
                debug!("None");
                Ok(())
            }
        }
    }

    pub(crate) async fn say_sync(&self, topic: &str, key: &str, val: &str) -> Result<()> {
        use crate::MqActor::*;
        match self {
            RedisMq(actor) => actor.publish(topic, key, val).await,
            RabbitMq(actor) => actor.publish(topic, key, val).await,
            _ => {
                debug!("None");
                Ok(())
            }
        }
    }
}
